package spring.cloud.tasks.tasks_manager.listener;

import com.alibaba.fastjson2.JSONObject;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import spring.cloud.tasks.common.Symbols;
import spring.cloud.tasks.common.nodepaths.*;
import spring.cloud.tasks.tasks_manager.schedule.ScheduleManager;
import spring.cloud.tasks.tasks_manager.schedule.TaskConfiguration;
import spring.cloud.tasks.tasks_manager.schedule.TaskInstanceInfo;
import spring.cloud.tasks.tasks_manager.task_item_assign.TaskItemAssignUtils;

import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;


/**
 * ZK事件分析
 * 命名空间任务分片重新分配的几率比较低。注意基于实例分片分配。如果出现命名空间重新分配，会造成大量数据处理异常。故该功能丢弃。
 * 任务启用+禁用的评率比较低。
 * 执行器上线和下线的几率比较高。
 * 执行器流控的比较低。
 * 那么只需要处理执行器频繁上线和下线的问题即可。尽量保证执行器不经常上线和下线操作。
 */
@Slf4j
@Setter
@Getter
public class TaskAndExecutorChangeListenerManager {

    protected ScheduleManager scheduleManager;

    public TaskAndExecutorChangeListenerManager(ScheduleManager scheduleManager) {
        this.scheduleManager = scheduleManager;
    }

    public void init() throws Exception {
        addTaskUpdateVersionNodeListener(scheduleManager, scheduleManager.curator.getCuratorFramework());
        addTaskInstanceStageStatusChangeNodeListener(scheduleManager, scheduleManager.curator.getCuratorFramework());
        //
        addDisableTaskNodeListener(scheduleManager.curator.getCuratorFramework());
        addTaskInstanceStopNodeListener(scheduleManager.curator.getCuratorFramework());
        addExecutorOfflineNodeListener(scheduleManager.curator.getCuratorFramework());
        addExecutorFollowControlNodeListener(scheduleManager.curator.getCuratorFramework());
    }


    private void addTaskUpdateVersionNodeListener(final ScheduleManager scheduleManager, CuratorFramework curatorFramework) throws Exception {

        /*子节点监听*/
        //子节点添加watcher
        //PathChildrenCache：监听数据节点的增删改，会触发事件
        //childData：设置缓存节点的数据状态
        final PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, TaskNodePaths.getROOT(), true);
        /**
         * StartMode：初始化方式
         * POST_INITIALIZED_EVENT：异步初始化。初始化后会触发事件
         * NORMAL：异步初始化
         * BUILD_INITIAL_CACHE：同步初始化
         * */
        pathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

        List<ChildData> childDataList = pathChildrenCache.getCurrentData();
        System.out.println("当前数据节点的子节点数据列表:");
        for (ChildData childData : childDataList) {
            String taskId = new String(childData.getData());
            System.out.println(taskId);
        }
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework1, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.INITIALIZED)) {
                    System.out.println("子节点初始化成功");

                } else if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED) || pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
                    String taskId = getTaskId(pathChildrenCacheEvent);
                    String taskVersion = new String(pathChildrenCacheEvent.getData().getData());
                    TaskConfiguration taskConfiguration = scheduleManager.taskConfigurationManager.getTaskConfiguration(taskId);
                    if (taskConfiguration == null) {
                        scheduleManager.taskConfigurationManager.taskIdToTaskConfigurationMap.put(taskVersion, taskConfiguration);
                    } else {
                        if (Long.parseLong(taskVersion) > taskConfiguration.getTaskVersion()) {
                            taskConfiguration.setTaskVersion(Long.parseLong(taskVersion));
                            scheduleManager.taskConfigurationManager.taskIdToTaskConfigurationMap.put(taskVersion, taskConfiguration);
                        }
                    }

                } else if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)) {
                    String taskId = getTaskId(pathChildrenCacheEvent);
                    TaskConfiguration taskConfiguration = scheduleManager.taskConfigurationManager.getTaskConfiguration(taskId);
                    if (taskConfiguration != null) {
                        taskConfiguration.setDelete(true);
                    }
                }
            }

            public String getTaskId(PathChildrenCacheEvent pathChildrenCacheEvent) {
                String path = pathChildrenCacheEvent.getData().getPath();
                String taskId = path.replace(TaskNodePaths.getROOT() + spring.cloud.tasks.common.Symbols.SLASH, "");
                return taskId;
            }
        });
    }

    private void addTaskInstanceStageStatusChangeNodeListener(ScheduleManager scheduleManager, CuratorFramework curatorFramework) throws Exception {
        // 参数  客户端，路径 ，缓存数据，是否压缩，线程池
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, TaskInstanceStageStatusChangeNodePaths.ROOT, true);
        //绑定监听器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {

                //监听子节点的变更,并且拿到变更后的数据
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                //判断类型是否是update
                if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(type)) {
                    //拿到数据
                    String path = pathChildrenCacheEvent.getData().getPath().replace(TaskInstanceStageStatusChangeNodePaths.ROOT + Symbols.SLASH, "");
                    String[] values = path.split(Symbols.RANGE);
                    String taskId = values[0];
                    String taskInstanceId = values[1];
                    byte[] bytes = pathChildrenCacheEvent.getData().getData();
                    long version = Long.parseLong(new String(bytes));
                    Thread thread = new Thread(new Runnable() {
                        @Override
                        public void run() {
                            //根据任务stage阶段的分片任务汇报的最新的状态进行处理
                            //比较状态更新的最新的版本，如果是当前版本比老的版本大则进行更新。如果小也更新，但是需要记录异常情况，备查。
                        }
                    });
                    thread.start();

                }
            }
        });
        //开启监听
        pathChildrenCache.start();
    }

    /**
     * 添加任务禁用监听器
     *
     * @param curatorFramework
     * @throws Exception
     */
    public void addDisableTaskNodeListener(CuratorFramework curatorFramework) throws Exception {
        // 参数  客户端，路径 ，缓存数据，是否压缩，线程池
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, EnabledTaskNodePaths.getRoot(), true);
        //绑定监听器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {

                //监听子节点的变更,并且拿到变更后的数据
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                //判断类型是否是update
                if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(type)) {
                    //拿到数据
                    byte[] bytes = pathChildrenCacheEvent.getData().getData();
                    if (bytes != null) {
                        JSONObject jsonObject = JSONObject.parseObject(new String(bytes));
                        String taskId = jsonObject.getString("taskId");
                        String taskType = jsonObject.getString("taskType");
                        String stateFullTask = jsonObject.getString("stateFullTask");
                        String taskInstanceId = jsonObject.getString("taskInstanceId");
                        Thread thread = new Thread(new Runnable() {
                            @Override
                            public void run() {
                                taskDisableOrTaskInstanceStopTaskItemAssign(taskId, taskType, stateFullTask, taskInstanceId);
                            }
                        });
                        thread.start();
                    }

                }
            }
        });
        //开启监听
        pathChildrenCache.start();
    }

    /**
     * 添加任务实例停止监听器
     *
     * @param curatorFramework
     * @throws Exception
     */
    public void addTaskInstanceStopNodeListener(CuratorFramework curatorFramework) throws Exception {
        // 参数  客户端，路径 ，缓存数据，是否压缩，线程池
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, TaskInstanceManageNodePaths.getTaskInstancesStopNodePath(), true);
        //绑定监听器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                //ZK标记为实例停止
                //监听子节点的变更,并且拿到变更后的数据
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                //判断类型是否是update
                if (PathChildrenCacheEvent.Type.CHILD_ADDED.equals(type)) {
                    //拿到数据
                    byte[] bytes = pathChildrenCacheEvent.getData().getData();
                    if (bytes != null) {
                        JSONObject jsonObject = JSONObject.parseObject(new String(bytes));
                        String taskId = jsonObject.getString("taskId");
                        String taskType = jsonObject.getString("taskType");
                        String stateFullTask = jsonObject.getString("stateFullTask");
                        //String taskInstanceId=jsonObject.getString("taskInstanceId");
                        Thread thread = new Thread(new Runnable() {
                            @Override
                            public void run() {
                                taskDisableOrTaskInstanceStopTaskItemAssign(taskId, taskType, stateFullTask, null);
                            }
                        });
                        thread.start();
                    }

                }
            }
        });
        //开启监听
        pathChildrenCache.start();
    }

    /**
     * 添加执行器下线监听器
     */
    public void addExecutorOfflineNodeListener(CuratorFramework curatorFramework) throws Exception {
        // 参数  客户端，路径 ，缓存数据，是否压缩，线程池
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, ExecutorNodePaths.getRoot(), true);
        //绑定监听器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {

                //监听子节点的变更,并且拿到变更后的数据
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                //判断类型是否是update
                if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(type)) {
                    //拿到数据
                    byte[] bytes = pathChildrenCacheEvent.getData().getData();
                    if (bytes != null) {
                        String executorId = new String(bytes);

                        Thread thread = new Thread(new Runnable() {
                            @Override
                            public void run() {
                                executorOfflineOrFollowControlTaskItemAssign(executorId, true);
                            }
                        });
                        thread.start();
                    }

                }
            }
        });
        //开启监听
        pathChildrenCache.start();
    }

    /**
     * 添加执行器下线监听器
     */
    public void addExecutorFollowControlNodeListener(CuratorFramework curatorFramework) throws Exception {
        // 参数  客户端，路径 ，缓存数据，是否压缩，线程池
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, ExecutorFlowControlNodePaths.getRoot(), true);
        //绑定监听器
        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {

                //监听子节点的变更,并且拿到变更后的数据
                PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
                //判断类型是否是update
                if (PathChildrenCacheEvent.Type.CHILD_REMOVED.equals(type)) {
                    //拿到数据
                    byte[] bytes = pathChildrenCacheEvent.getData().getData();
                    if (bytes != null) {
                        String executorId = new String(bytes);
                        executorOfflineOrFollowControlTaskItemAssign(executorId, false);

                    }

                }
            }
        });
        //开启监听
        pathChildrenCache.start();
    }

    public void taskDisableOrTaskInstanceStopTaskItemAssign(String taskId, String taskType, String stateFullTask, String taskInstanceId) {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                Set<String> taskInstanceIdSet = new HashSet<>();//当taskInstanceId==null时使用
                if (taskInstanceId != null) {
                    TaskInstanceInfo taskInstanceInfo = TaskItemAssignUtils.getTaskInstanceInfo(scheduleManager, taskId, taskType, stateFullTask, taskInstanceId);
                    if (taskInstanceInfo != null) {
                        taskInstanceInfo.setTryCancel(true);
                        taskInstanceInfo.setTryCancelDate(new Date());
                        //
                        taskInstanceIdSet.add(taskInstanceId);
                    }
                } else {
                    List<TaskInstanceInfo> taskInstanceInfoList = TaskItemAssignUtils.getTaskInstanceInfoList(scheduleManager, taskId, taskType, stateFullTask);
                    if (taskInstanceInfoList != null) {
                        for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                            taskInstanceInfo.setTryCancel(true);
                            taskInstanceInfo.setTryCancelDate(new Date());
                            //
                            taskInstanceIdSet.add(taskInstanceInfo.getTaskInstanceId());
                        }

                    }
                }

                //尝试一次实例取消的调度处理，同时执行器也进行实例取消。
                TaskDisableOrTaskInstanceStopTaskItemAssign taskDisableOrTaskInstanceStopTaskItemAssign =
                        new TaskDisableOrTaskInstanceStopTaskItemAssign();
                try {
                    Lock lock = scheduleManager.taskItemAssignLock;
                    boolean tryLock = lock.tryLock(15, TimeUnit.SECONDS);
                    if (tryLock) {
                        try {
                            //
                            Set<String> taskInstanceIdSet2 = taskDisableOrTaskInstanceStopTaskItemAssign.taskItemAssign(scheduleManager, taskId, taskInstanceId);
                            taskInstanceIdSet.addAll(taskInstanceIdSet2);
                            taskDisableOrTaskInstanceStopTaskItemAssign.taskInstanceStausUpdate(scheduleManager, taskId, taskType, stateFullTask, taskInstanceIdSet);
                        } finally {
                            lock.unlock();
                        }
                    } else {
                        taskDisableOrTaskInstanceStopTaskItemAssign.taskInstanceStausUpdate(scheduleManager, taskId, taskType, stateFullTask, taskInstanceIdSet);
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        });
        thread.start();
    }

    public void executorOfflineOrFollowControlTaskItemAssign(String executorId, boolean offlineOrFollowControl) {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {

                try {
                    Lock lock = scheduleManager.taskItemAssignLock;
                    boolean tryLock = lock.tryLock(15, TimeUnit.SECONDS);
                    if (tryLock) {
                        try {
                            ExecutorOfflineOrFollowControlTaskItemAssign executorOfflineOrFollowControlTaskItemAssign = new ExecutorOfflineOrFollowControlTaskItemAssign();
                            executorOfflineOrFollowControlTaskItemAssign.taskItemAssign(scheduleManager, executorId, true);
                        } finally {
                            lock.unlock();
                        }
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        });
        thread.start();
    }
}
